use std::time::{Duration, Instant};

use actix_ws::Message;
use futures_util::{
    future::{select, Either},
    StreamExt as _,
};
use tokio::{pin, sync::mpsc, time::interval};

use crate::ws::{
    resp::{ConnId, MessageResp, SystemResp},
    server::ChatServerHandle,
};

/// 发送心跳信号的频率
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);

/// 客户端未响应超时时长
const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);

/// 处理回显从客户端接收的文本和二进制消息，响应 ping 消息，进行监视连接及运行状况，以检测网络问题并释放资源
pub async fn chat_ws(
    chat_server: ChatServerHandle,
    mut session: actix_ws::Session,
    mut msg_stream: actix_ws::MessageStream,
) {
    let mut name = None;
    let mut last_heartbeat = Instant::now();
    let mut interval = interval(HEARTBEAT_INTERVAL);

    let (conn_tx, mut conn_rx) = mpsc::unbounded_channel();

    // unwrap: 在HTTP服务器之前不会删除聊天服务器
    let conn_id = chat_server.connect(conn_tx).await;

    log::info!("connected connId: {conn_id}");

    let close_reason = loop {
        let tick = interval.tick();
        pin!(tick);

        let msg_rx = conn_rx.recv();
        pin!(msg_rx);

        let messages = select(msg_stream.next(), msg_rx);
        pin!(messages);

        match select(messages, tick).await {
            // 从客户端接收的命令和消息
            Either::Left((Either::Left((Some(Ok(msg)), _)), _)) => {
                log::debug!("msg: {msg:?}");

                match msg {
                    Message::Ping(bytes) => {
                        log::warn!("ping {:?}", &bytes);
                        last_heartbeat = Instant::now();
                        session.pong(&bytes).await.unwrap();
                    }

                    Message::Pong(_) => {
                        last_heartbeat = Instant::now();
                    }

                    Message::Text(text) => {
                        process_text_msg(&chat_server, &mut session, &text, conn_id, &mut name)
                            .await;
                    }

                    Message::Binary(_bin) => {
                        log::warn!("unexpected binary message");
                    }

                    Message::Close(reason) => break reason,

                    _ => {
                        break None;
                    }
                }
            }

            // 客户端 WebSocket 流错误
            Either::Left((Either::Left((Some(Err(err)), _)), _)) => {
                log::error!("{}", err);
                break None;
            }

            // 客户端 WebSocket 流已结束
            Either::Left((Either::Left((None, _)), _)) => break None,

            // 从其他聊天室参与者收到的聊天信息
            Either::Left((Either::Right((Some(chat_msg), _)), _)) => {
                session.text(chat_msg).await.unwrap();
            }

            // 所有消息发送者被删除
            Either::Left((Either::Right((None, _)), _)) => unreachable!(
                "all connection message senders were dropped; chat server may have panicked"
            ),

            // 心跳循环
            Either::Right((_inst, _)) => {
                // 如果没收到心跳 ping/pong, 将会关闭连接
                if Instant::now().duration_since(last_heartbeat) > CLIENT_TIMEOUT {
                    log::info!(
                        "client has not sent heartbeat in over {CLIENT_TIMEOUT:?}; disconnecting"
                    );
                    break None;
                }

                // 发送心跳
                let _ = session.ping(b"").await;
            }
        };
    };

    chat_server.disconnect(conn_id);

    // 尝试正常关闭连接
    let _ = session.close(close_reason).await;
}

async fn process_text_msg(
    chat_server: &ChatServerHandle,
    session: &mut actix_ws::Session,
    text: &str,
    conn: ConnId,
    name: &mut Option<String>,
) {
    let msg = text.trim();

    // 分析命令
    if msg.starts_with('/') {
        let mut cmd_args = msg.splitn(2, ' ');

        log::info!("ws command msg {msg}");

        // unwrap: 已确保字符串长度不为0
        match cmd_args.next().unwrap() {
            // 获取所有房间列表
            "/rooms" => {
                let rooms = chat_server.list_rooms().await;
                let rooms_len = rooms.len();
                let room_resp = SystemResp {
                    event: "system".to_owned(),
                    kind: "rooms".to_owned(),
                    data: rooms,
                    number: rooms_len,
                };
                // for room in rooms {
                //     session.text(room).await.unwrap();
                // }
                let res = serde_json::to_string(&room_resp).unwrap();
                session.text(res).await.unwrap();
            }

            // 获取所有用户列表
            "/ids" => {
                let ss = chat_server.list_connids().await;
                let ss_len = ss.len();
                let ids_resp = SystemResp {
                    event: "system".to_owned(),
                    kind: "ids".to_owned(),
                    data: ss,
                    number: ss_len,
                };
                let res = serde_json::to_string(&ids_resp).unwrap();
                session.text(res).await.unwrap();
            }

            // 获取所有用户信息列表
            "/users" => {
                let users = chat_server.list_users().await;
                let users_len = users.len();
                let users_resp = SystemResp {
                    event: "system".to_owned(),
                    kind: "users".to_owned(),
                    data: users,
                    number: users_len,
                };
                let res = serde_json::to_string(&users_resp).unwrap();
                session.text(res).await.unwrap();
            }

            // 加入某个房间
            "/join" => match cmd_args.next() {
                Some(room) => {
                    log::info!("conn {conn}: joining room {room}");
                    chat_server.join_room(conn, room).await;
                    session.text(format!("joined {room}")).await.unwrap();
                }
                None => {
                    session.text("!!! room name is required").await.unwrap();
                }
            },

            // 改变当前名称
            "/name" => match cmd_args.next() {
                Some(new_name) => {
                    log::info!("conn {conn} setting name to: {new_name}");
                    name.replace(new_name.to_owned());
                    chat_server.set_user(conn, new_name.to_owned()).await;
                    session
                        .text(format!("changed name {new_name}"))
                        .await
                        .unwrap();
                }
                None => {
                    session.text("!!! name is required").await.unwrap();
                }
            },

            _ => {
                session
                    .text(format!("!!! unknown command: {msg}"))
                    .await
                    .unwrap();
            }
        }
    } else {
        // 消息前加上名称
        // let msg = match name {
        //     Some(ref name) => format!("{name}: {msg}"),
        //     None => msg.to_owned(),
        // };
        // 设置发送者名称
        let name = match name {
            Some(ref name) => name,
            None => "None",
        };
        // 转换数据形成 JSON 字符串
        let msg_resp = MessageResp {
            event: "message".to_owned(),
            from: name.to_owned(),
            to: "room".to_owned(),
            data: msg,
        };
        let res = serde_json::to_string(&msg_resp).unwrap();
        chat_server.send_message(conn, res).await
    }
}
